package com.atguigu.chapter11;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * @ClassName: Flink14_Function_Scalar
 * @Description:
 * @Author: kele
 * @Date: 2021/4/14 0:09
 *
 * 表值聚合函数：
 *
 *
 *
 * 需求：选出每个id中最大的前两名的vc
 *          --前  ： 需要聚合
 *          --两名 ： 需要炸裂（一行变多行）
 **/
public class Flink17_Function__TableAggregate {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        DataStreamSource<WaterSensor> ds = env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_2", 6000L, 60));

        Table table = tenv.fromDataStream(ds);

        /**
         * Table API ：内联方式
         */

      /*  table
                .groupBy($("id"))
                .flatAggregate(call(TopTwo.class,$("vc")).as("nnvc","nnum")  )
                .select($("id"),$("nnvc"),$("nnum"))
                .execute()
                .print();*/


        /**
         *
         * Table API：注册
         */
        tenv.createTemporaryFunction("topNUM",TopTwo.class);

        table
                .groupBy($("id"))
                .flatAggregate(call("topNUM",$("vc")).as("nnvc","nnum")  )
                .select($("id"),$("nnvc"),$("nnum"))
                .execute()
                .print();



        /**
         * SQL : 不支持
         */

    }

    public static class Top2Acc {
        public Integer first = Integer.MIN_VALUE;
        public Integer second = Integer.MIN_VALUE;
    }

    @FunctionHint(output = @DataTypeHint("Row<nvc Integer,num Integer>"))
    public static class TopTwo extends TableAggregateFunction<Row,Top2Acc>{

        @Override
        public Top2Acc createAccumulator() {
            return new Top2Acc();
        }

        /**
         * 计算中间计算结果
         * @param acc
         * @param value
         */
        public void accumulate(Top2Acc acc, Integer value){

            if(value > acc.first){
                acc.second = acc.first;
                acc.first = value;
            }else if(value > acc.second){
                acc.second = value;
            }
        }


        /**
         * 返回最终结果
         */
        public void emitValue(Top2Acc acc, Collector<Row> out){

            if(acc.first != Integer.MIN_VALUE){

                Row row = Row.of(acc.first, 1);
                out.collect(row);
            }

            if(acc.second != Integer.MIN_VALUE){
                Row row = Row.of(acc.second, 2);
                out.collect(row);
            }
        }

    }



}
